"""
Case Type   : 基础功能
Case Name   : 发布订阅--pg_replication_origin_status
Create At   : 2022/1/26
Owner       : opentestcase026
Description :
    1.创建发布订阅
    2.查看 pg_replication_origin_status
    3.修改订阅端信息
    4.查询系统表内容
Expect      :
    1.成功
    2.类型与资料一致
    3.成功
    4.内容正确
History     :
    Modified by opentestcase004 2022/8/11:优化用例名打印方式
    Modified by opentestcase012 2023/5/9:优化用例,等待数据同步
    Modified by opentestcase012 2023/5/15:优化用例断言
    Modified by opentestcase024 2023/6/8:优化用例，准确验证视图查询结果
"""
import unittest
import os
from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.subname = "sub_case131"
        self.pubname = "pub_case131"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'

        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')}" \
            f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        self.tb_name = "tb_pub_sub_case131"
        text = '--step:预置条件,修改pg_hba expect:成功'
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '--step1:创建发布订阅 expect:成功--'
        self.log.info(text)
        sql = f"create table {self.tb_name}(i int primary key);" \
            f"CREATE PUBLICATION {self.pubname}  FOR all tables;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql1 = f"create table {self.tb_name}(i int primary key);"
        sql = f"CREATE SUBSCRIPTION {self.subname} CONNECTION " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"PUBLICATION {self.pubname};"
        result = self.commsh_sub.execut_db_sql(sql1+sql,
                                               self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.create_sub_succ_msg,
                      result, '执行失败:' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS,
                      result, '执行失败:' + text)

        text = '--step2:查看 pg_replication_origin_status  ' \
               'expect:类型与资料一致--'
        self.log.info(text)
        sql = "\d+ pg_replication_origin_status"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('local_id    | oid  |           | plain',
                      result, '执行失败:' + text)
        self.assertIn('external_id | text |           | extended',
                      result, '执行失败:' + text)
        self.assertIn('remote_lsn  | text |           | extended',
                      result, '执行失败:' + text)
        self.assertIn('local_lsn   | text |           | extended',
                      result, '执行失败:' + text)

        text = '--step3:修改数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name} values(1);" \
            f"select pg_current_xlog_location();"
        result = self.commsh_pub.execut_db_sql(sql,
                                               self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      result, '执行失败:' + text)
        new_lsn = result.splitlines()[-2].strip()
        self.log.info(f'修改数据后lsn值为：{new_lsn}')

        self.log.info("----等待订阅端数据同步，超时时间为30s----")
        self.commsh_pub.check_sub_consistency(self.subname,
                                              self.user_param_pub, 30)

        sql = f"select * from pg_replication_origin_status;" \
              "select pg_current_xlog_location();"
        result = self.commsh_pub.execut_db_sql(sql,
                                               self.user_param_pub)
        self.log.info(result)

        text = '--step4:查询系统表内容  expect:内容正确--'
        self.log.info(text)
        sql = "select * from pg_replication_origin_status;" \
              "select pg_current_xlog_location();"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        remote_lsn = result.splitlines()[2].split('|')[2].strip()
        local_lsn = result.splitlines()[2].split('|')[3].strip()
        self.log.info(f'pg_replication_origin_status视图remote_lsn为：'
                      f'{remote_lsn}')
        self.log.info(f'pg_replication_origin_status视图local_lsn为：'
                      f'{local_lsn}')
        text1 = '预期1：执行修改操作事务后,pg_replication_origin_status视图' \
                'remote_lsn等于事务执行完成的lsn'
        self.log.info(text1)
        self.assertEqual(new_lsn, remote_lsn, '执行失败:' + text + text1)
        text2 = '预期2：执行修改操作事务后,pg_replication_origin_status视图local_lsn' \
                '小于等于当前pg_current_xlog_location'
        self.log.info(text2)
        self.assertLessEqual(local_lsn, result.splitlines()[-2].strip(),
                             '执行失败:' + text + text2)

    def tearDown(self):
            self.log.info('------------this is tearDown-------------')
            text = '--清理环境--'
            self.log.info(text)
            sql = f"DROP SUBSCRIPTION  {self.subname};" \
                  f"drop table {self.tb_name};"
            drop_sub_result = self.commsh_sub.execut_db_sql(
                sql, self.user_param_sub,
                None, macro.DB_ENV_PATH_REMOTE1)
            self.log.info(drop_sub_result)
            sql = f"DROP PUBLICATION if exists {self.pubname};" \
                f"drop table {self.tb_name};"
            drop_pub_result = self.commsh_pub.execut_db_sql(
                sql, sql_type=self.user_param_pub)
            self.log.info(drop_pub_result)
            cmd = f"mv " \
                f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')} "\
                f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} "
            self.log.info(cmd)
            result = self.pri_userdb_pub.sh(cmd).result()
            self.log.info(result)
            path = os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')
            cmd = f"mv " \
                f"{os.path.join(self.parent_path_sub, 'pg_hba.conf')} "\
                f"{path} "
            self.log.info(cmd)
            result = self.pri_userdb_sub.sh(cmd).result()
            self.log.info(result)
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'wal_level={self.wal_level}')
            self.assertTrue(result, '执行失败:' + text)
            self.commsh_pub.restart_db_cluster(True)
            self.commsh_sub.restart_db_cluster(True,
                                               macro.DB_ENV_PATH_REMOTE1)
            self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                          '执行失败' + text)
            self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                          '执行失败' + text)
            self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_sub_result,
                          '执行失败' + text)
            self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_pub_result,
                          '执行失败' + text)
            self.log.info(f"-----{os.path.basename(__file__)} end-----")
